package com.bjmashibing.system.io;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

public class SocketMultiplexingSingleThreadv1 {

    private ServerSocketChannel server = null;
    private Selector selector = null;//linux多路复用器（select poll epoll kqueue）
    int port = 9090;

    public void initServer() {
        try {
            server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.bind(new InetSocketAddress(port));

            //如果在epoll模型下 open() 完成了epoll create --> fd3 内核中创建红黑树保存fds
            //cat /proc/sys/fs/epoll/max_user_watches查看内核中user名义下的红黑树的个数
            selector = Selector.open();  //  select  poll  *epoll 优先选择:epoll 但是可以-D修正

            //server等于listen状态的fd4
            /*
            * 如果：
            * select, poll: jvm里开辟一个数组 fd 放进去
            * epoll: epoll_ctl(fd3,ADD,fd4,EPOLLIN)
            * */
            server.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void start() {
        initServer();
        System.out.println("服务器启动了。。。。。");
        try {
            while (true) {
                Set<SelectionKey> keys = selector.keys();
//                System.out.println(keys.size()+"   size");
                /*
                * 1、调用多路复用器（select，poll or epoll (epoll_wait)）
                *
                * select()方法的含义：
                * 1、select，poll多路复用器其实是内核的select(fd4[fds])  poll(fd4)  系统调用，将进程内存中的fds作为参数进行系统调用
                * 2、epoll:其实是内核的epoll.wait()  不会传数据
                *    参数可以带时间：0：阻塞，有时间表示设置一个超时
                *     select.wakeup()  结果容易返回0
                * */

                /*
                * register类似懒加载的感觉：
                * 其实是触碰到了selector.select()调用的时候触发了epoll_ctl调用
                * */
                while (selector.select(500) > 0) {//他只会返回他调起时候已经注册的fd
                    //NIO 自己对着每一个fd进行系统调用，会浪费资源，这里一次select方法就知道具体哪些fd可以R/W
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();//返回的有状态的fd集合
                    Iterator<SelectionKey> iter = selectionKeys.iterator();
                    //管你啥多路复用器，只能给我状态
                    //socket: 分为两类 listen 和 通信 R/W
                    while (iter.hasNext()) {
                        SelectionKey key = iter.next();
                        iter.remove();//set 不移除会重复循环处理
                        if (key.isAcceptable()) {
                            //接受一个新的连接
                            //accept接受连接并且返回新连接的fd
                            //select,poll 因为他们内核没有空间，那么在jvm中保存和前面的fd4那个listen的一起
                            //epoll:  通过epoll_ctl把新的客户端fd注册到内核空间
                            acceptHandler(key);
                        } else if (key.isReadable()) {
                            key.cancel();//在多路复用器里把key cancel了
                            readHandler(key);//数据没有读完，read事件会重复触发
                            //在当前线程，这个方法可能会阻塞，可能当前读取的数据的业务逻辑处理时间比较长
                            //IO Threads  (IO和处理上解耦) IO是单独的一件事
                        }else if(key.isWritable()){//send-queue是空的
                            key.cancel();//epoll: epoll_ctl(fd3,DELETE,fd  删除了红黑树里的fd
                            writeHandler(key);
                        }
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void acceptHandler(SelectionKey key) {
        try {
            ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
            SocketChannel client = ssc.accept();//目的是调用accept接收客户端 fd7
            client.configureBlocking(false);

            ByteBuffer buffer = ByteBuffer.allocate(8192);
            /*
            *  select, poll: jvm里开辟一个数组 fd7 放进去
            *  epoll: epoll_ctl(fd3,ADD,fd7,EPOLLIN)
            * */
            client.register(selector, SelectionKey.OP_READ, buffer);
            System.out.println("-------------------------------------------");
            System.out.println("新客户端：" + client.getRemoteAddress());
            System.out.println("-------------------------------------------");

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void writeHandler(SelectionKey key){
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("write handler...");
                SocketChannel client = (SocketChannel) key.channel();
                ByteBuffer buffer = (ByteBuffer) key.attachment();
                buffer.flip();
                while(buffer.hasRemaining()){
                    try {
                        client.write(buffer);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                buffer.clear();
                key.cancel();
                try {
                    client.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public void readHandler(SelectionKey key) {
        new Thread(() ->{
            System.out.println("read handler...");
            SocketChannel client = (SocketChannel) key.channel();
            ByteBuffer buffer = (ByteBuffer) key.attachment();
            buffer.clear();
            int read = 0;
            try {
                while (true) {
                    read = client.read(buffer);
                    if (read > 0) {

                        client.register(key.selector(), SelectionKey.OP_WRITE,buffer);
                        //关心 send-queue是不是有空间

//                    buffer.flip();
//                    while (buffer.hasRemaining()) {
//                        byte[] bytes = new byte[buffer.limit()];
//                        buffer.get(bytes);
//                        System.out.println(new String(bytes));
//                        client.write(buffer);
//                        System.out.println(buffer);
//                    }
//                    buffer.clear();
//                    System.out.println(buffer);
                    } else if (read == 0) {
                        System.out.println("000000");
                        break;
                    } else {//-1表示对方已经断开连接
                        client.close();
                        System.out.println("close");
                        break;
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
    }

    public static void main(String[] args) {
        SocketMultiplexingSingleThreadv1 service = new SocketMultiplexingSingleThreadv1();
        service.start();
    }
}
